package service

import (
	"context"
	"time"

	"task/config"
	"task/driver"
	"task/ecode"
	"task/models"

	log "github.com/sirupsen/logrus"
)

type DispatchEvent interface {
	Prepare(context.Context, *models.DispatchEvent) error
	Compute(context.Context, *models.DispatchEvent) error
	After(context.Context, *models.DispatchEvent) error
}

var (
	DispatchEventMap = make(map[string]DispatchEvent)
	DagMap           = make(map[int]*models.Dag)
)

// InitDispatchEvent init dispatch event.
func InitDispatchEvent() {
	DispatchEventMap[models.DispatchEventResourceAdd] = ResourceAddEvent{}
	DispatchEventMap[models.DispatchEventProcessorDone] = ProcessorDoneEvent{}
	DispatchEventMap[models.DispatchEventBusinessCallback] = BusinessCallback{}
}

// InitDagMap init dag map.
func InitDagMap() {
	getDagMap()
	go func() {
		for {
			getDagMap()
			time.Sleep(time.Duration(5) * time.Minute)
		}
	}()
}

// getDagMap get dag map.
func getDagMap() {
	log.Infof("get dag map...")
	var (
		dags []*models.Dag
		err  error
	)
	if dags, err = driver.ListDag(); err != nil {
		log.Panic("getDagMap error: %v", err)
	}
	for _, dag := range dags {
		DagMap[dag.Id] = dag
	}
	return
}

// getDispatchEvent get dispatch event.
func getDispatchEvent(name string) (DispatchEvent, error) {
	var (
		event DispatchEvent
		ok    bool
	)
	if event, ok = DispatchEventMap[name]; !ok {
		return nil, ecode.DispatchEventNotFound
	}
	return event, nil
}

// ResourceAddEvent resource add event.
// 调度事件:资源添加
type ResourceAddEvent struct {
}

func (h ResourceAddEvent) Prepare(_ context.Context, event *models.DispatchEvent) error {
	var (
		ok  bool
		err error
	)
	// 检查资源
	if event.Resource, err = driver.GetResource(event.ResourceId); err != nil {
		return err
	}
	if event.Resource == nil {
		return ecode.ResourceNotFound
	}
	// 检查DAG
	if event.Dag, ok = DagMap[event.DagId]; ok != true {
		return err
	}
	if event.Dag == nil {
		return ecode.DagNotFound
	}
	return nil
}

func (h ResourceAddEvent) Compute(_ context.Context, event *models.DispatchEvent) error {
	var err error
	// 生成DAG图
	if event.Graph, err = generateGraph(event.Dag.Config); err != nil {
		return err
	}
	// 计算根处理单元
	for _, node := range event.Graph {
		if node.Parents == nil {
			event.ExecutorList = append(event.ExecutorList, int64(node.ProcessorId))
			// 设置处理状态为等待执行
			if err = driver.AddResourceProcessState(event.ResourceId, node.ProcessorId, models.ProcessStateReady, 0, ""); err != nil {
				return err
			}
		}
	}
	return nil
}

func (h ResourceAddEvent) After(ctx context.Context, event *models.DispatchEvent) error {
	var (
		executeEvents []*models.ProcessEvent
		err           error
	)
	if len(event.ExecutorList) == 0 {
		return nil
	}
	// 发送执行消息
	for _, processorId := range event.ExecutorList {
		executeEvents = append(executeEvents, &models.ProcessEvent{
			ResourceId:  event.ResourceId,
			ProcessorId: int(processorId),
			ProcessType: models.ProcessTypeExecute,
		})
	}
	if err = driver.SendExecuteEventMsg(ctx, executeEvents); err != nil {
		return err
	}
	return nil
}

// ProcessorDoneEvent processor done event.
// 调度事件:处理完成
type ProcessorDoneEvent struct {
}

func (h ProcessorDoneEvent) Prepare(_ context.Context, event *models.DispatchEvent) error {
	var (
		ok  bool
		err error
	)
	// 检查资源
	if event.Resource, err = driver.GetResource(event.ResourceId); err != nil {
		return err
	}
	if event.Resource == nil {
		return ecode.ResourceNotFound
	}
	// 检查DAG
	if event.Dag, ok = DagMap[event.DagId]; ok != true {
		return err
	}
	if event.Dag == nil {
		return ecode.DagNotFound
	}
	// 检查处理状态
	if event.ResourceProcessState, err = driver.GetResourceProcessState(event.ResourceId, event.ProcessorId); err != nil {
		return err
	}
	if event.ResourceProcessState == nil {
		return ecode.ProcessStateNotFound
	}
	// 最大失败重试次数
	if event.ResourceProcessState.ProcessCnt > config.Conf.TaskDispatch.RetryCnt {
		log.Errorf("Resource(%d) processor(%d) process count(%d) over max retry times", event.ProcessorId,
			event.ProcessorId, event.ResourceProcessState.ProcessCnt)
		return ecode.DispatchMaxRetryTimes
	}
	return nil
}

func (h ProcessorDoneEvent) Compute(_ context.Context, event *models.DispatchEvent) error {
	var (
		children []*models.Node
		err      error
	)
	// 失败任务重试
	if event.ResourceProcessState.ProcessState == models.ProcessStateFail {
		event.ExecutorList = append(event.ExecutorList, int64(event.ProcessorId))
		// 设置处理状态为等待执行
		if err = driver.UpdateResourceProcessState(event.ResourceProcessState.Id, models.ProcessStateReady,
			event.ResourceProcessState.ProcessCnt, ""); err != nil {
			return err
		}
		return nil
	}
	// 生成DAG图
	if event.Graph, err = generateGraph(event.Dag.Config); err != nil {
		return err
	}
	// 寻找下游节点
	if children = event.Graph[event.ProcessorId].Children; children == nil {
		// 没有下游 说明该处理单元是末节点 直接返回
		return nil
	}
	// 每个下游节点判断其所有上游节点是否都执行成功
	for _, childNode := range children {
		ready := true
		for _, node := range childNode.Parents {
			// 自身节点跳过
			if node.ProcessorId == event.ProcessorId {
				continue
			}
			var processState *models.ResourceProcessState
			if processState, err = driver.GetResourceProcessState(event.ResourceId, node.ProcessorId); err != nil {
				return err
			}
			// 上游节点 处理状态不存在(等同于初始状态) 或者 存在但不等于处理成功 都不能调度该下游节点
			if processState == nil || processState.ProcessState != models.ProcessStateSuccess {
				ready = false
				break
			}
		}
		if !ready {
			continue
		}
		var processState *models.ResourceProcessState
		if processState, err = driver.GetResourceProcessState(event.ResourceId, childNode.ProcessorId); err != nil {
			return err
		}
		// 下游节点 处理状态不存在(等同于初始状态) 创建处理状态设置为ready
		if processState == nil {
			if err = driver.AddResourceProcessState(event.ResourceId, childNode.ProcessorId, models.ProcessStateReady,
				0, ""); err != nil {
				return err
			}
			event.ExecutorList = append(event.ExecutorList, int64(childNode.ProcessorId))
		}
	}
	return nil
}

func (h ProcessorDoneEvent) After(ctx context.Context, event *models.DispatchEvent) error {
	var (
		executeEvents []*models.ProcessEvent
		err           error
	)
	if len(event.ExecutorList) == 0 {
		return nil
	}
	// 发送执行消息
	for _, processorId := range event.ExecutorList {
		executeEvents = append(executeEvents, &models.ProcessEvent{
			ResourceId:  event.ResourceId,
			ProcessorId: int(processorId),
			ProcessType: models.ProcessTypeExecute,
		})
	}
	if err = driver.SendExecuteEventMsg(ctx, executeEvents); err != nil {
		return err
	}
	return nil
}

// BusinessCallback business callback.
// 调度事件:业务回调
type BusinessCallback struct {
}

func (h BusinessCallback) Prepare(_ context.Context, event *models.DispatchEvent) error {
	var err error
	// 检查资源
	if event.Resource, err = driver.GetResource(event.ResourceId); err != nil {
		return err
	}
	if event.Resource == nil {
		return ecode.ResourceNotFound
	}
	// 检查处理状态
	if event.ResourceProcessState, err = driver.GetResourceProcessState(event.ResourceId, event.ProcessorId); err != nil {
		return err
	}
	if event.ResourceProcessState == nil {
		return ecode.ProcessStateNotFound
	}
	if event.ResourceProcessState.ProcessState != models.ProcessStateWait {
		return ecode.ProcessStateWrong
	}
	return nil
}

func (h BusinessCallback) Compute(_ context.Context, event *models.DispatchEvent) error {
	var err error
	// 设置处理状态为等待执行
	if err = driver.UpdateResourceProcessState(event.ResourceProcessState.Id, models.ProcessStateReady,
		event.ResourceProcessState.ProcessCnt, ""); err != nil {
		return err
	}
	event.ExecutorList = append(event.ExecutorList, int64(event.ProcessorId))
	return nil
}

func (h BusinessCallback) After(ctx context.Context, event *models.DispatchEvent) error {
	var (
		executeEvents []*models.ProcessEvent
		err           error
	)
	if len(event.ExecutorList) == 0 {
		return nil
	}
	// 发送执行消息
	for _, processorId := range event.ExecutorList {
		executeEvents = append(executeEvents, &models.ProcessEvent{
			ResourceId:  event.ResourceId,
			ProcessorId: int(processorId),
			ProcessType: models.ProcessTypeCallback,
			Result:      event.Result,
		})
	}
	if err = driver.SendExecuteEventMsg(ctx, executeEvents); err != nil {
		return err
	}
	return nil
}
